// Copyright(C) 2023 InfiniFlow, Inc. All rights reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     https://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

module;

#include <cassert>

module infinity_core:new_txn_index.impl;

import :new_txn;
import :new_txn_manager;
import :kv_store;
import :chunk_index_meta;
import :segment_index_meta;
import :table_index_meta;
import :catalog_meta;
import :db_meta;
import :table_meta;
import :segment_meta;
import :block_meta;
import :column_meta;
import :logger;
import :infinity_context;
import :buffer_manager;
import :infinity_exception;
import :buffer_obj;
import :mem_index;
import :wal_entry;
import :secondary_index_in_mem;
import :secondary_index_data;
import :default_values;
import :ivf_index_data_in_mem;
import :ivf_index_data;
import :memory_indexer;
import :index_full_text;
import :column_index_reader;
import :column_index_merger;
import :hnsw_handler;
import :bmp_handler;
import :index_hnsw;
import :index_bmp;
import :emvb_index_in_mem;
import :emvb_index;
import :meta_key;
import :data_access_state;
import :hnsw_util;
import :bmp_util;
import :defer_op;
import :base_txn_store;
import :kv_code;
import :buffer_handle;
import :bg_task;
import :mem_index_appender;
import :txn_context;
import :kv_utility;
import :dump_index_process;
import :persistence_manager;
import :persist_result_handler;

import std;
import third_party;

import statement_common;
import column_def;
import internal_types;
import create_index_info;
import embedding_info;

namespace infinity {

Status NewTxn::DumpMemIndex(const std::string &db_name, const std::string &table_name, const std::string &index_name) {
    Status status;

    std::shared_ptr<DBMeta> db_meta;
    std::shared_ptr<TableMeta> table_meta;
    std::shared_ptr<TableIndexMeta> table_index_meta;
    std::string table_key;
    std::string index_key;
    status = GetTableIndexMeta(db_name, table_name, index_name, db_meta, table_meta, table_index_meta, &table_key, &index_key);
    if (!status.ok()) {
        return status;
    }
    std::vector<SegmentID> *segment_ids_ptr = nullptr;
    std::tie(segment_ids_ptr, status) = table_index_meta->GetSegmentIndexIDs1();
    if (!status.ok()) {
        return status;
    }

    // Put the data into local txn store
    if (base_txn_store_ != nullptr) {
        return Status::UnexpectedError("txn store is not null");
    }
    base_txn_store_ = std::make_shared<DumpMemIndexTxnStore>();
    DumpMemIndexTxnStore *txn_store = static_cast<DumpMemIndexTxnStore *>(base_txn_store_.get());
    txn_store->db_name_ = db_name;
    txn_store->db_id_str_ = table_index_meta->table_meta().db_id_str();
    txn_store->table_name_ = table_name;
    txn_store->table_id_str_ = table_index_meta->table_meta().table_id_str();
    txn_store->index_name_ = index_name;
    txn_store->index_id_str_ = table_index_meta->index_id_str();
    txn_store->index_id_ = std::stoull(txn_store->index_id_str_);
    txn_store->segment_ids_ = *segment_ids_ptr;
    txn_store->table_key_ = table_key;

    for (SegmentID segment_id : *segment_ids_ptr) {
        SegmentIndexMeta segment_index_meta(segment_id, *table_index_meta);

        std::shared_ptr<MemIndex> mem_index = segment_index_meta.GetMemIndex();
        if (mem_index == nullptr || mem_index->IsDumping() || (mem_index->GetBaseMemIndex() == nullptr && mem_index->GetEMVBIndex() == nullptr)) {
            continue;
        }
        mem_index->SetIsDumping(true);

        ChunkIndexMetaInfo chunk_index_meta_info;
        if (mem_index->GetBaseMemIndex() != nullptr) {
            chunk_index_meta_info = mem_index->GetBaseMemIndex()->GetChunkIndexMetaInfo();
        } else if (mem_index->GetEMVBIndex() != nullptr) {
            chunk_index_meta_info = mem_index->GetEMVBIndex()->GetChunkIndexMetaInfo();
        } else {
            return Status::UnexpectedError("Invalid mem index.");
        }

        ChunkID chunk_id = 0;
        std::tie(chunk_id, status) = segment_index_meta.GetAndSetNextChunkID();
        if (!status.ok()) {
            return status;
        }

        std::vector<WalChunkIndexInfo> chunk_infos;
        chunk_infos.emplace_back(chunk_index_meta_info, chunk_id);
        txn_store->chunk_infos_in_segments_.emplace(segment_id, chunk_infos);

        // Dump Mem Index
        status = this->DumpSegmentMemIndex(segment_index_meta, chunk_id);
        if (!status.ok() && status.code() != ErrorCode::kEmptyMemIndex) {
            return status;
        }
    }

    if (txn_store->chunk_infos_in_segments_.empty()) {
        base_txn_store_ = nullptr; // No mem index to dump.
    }

    return Status::OK();
}

Status NewTxn::DumpMemIndex(const std::string &db_name,
                            const std::string &table_name,
                            const std::string &index_name,
                            SegmentID segment_id,
                            RowID begin_row_id) {
    Status status;

    std::shared_ptr<DBMeta> db_meta;
    std::shared_ptr<TableMeta> table_meta;
    std::shared_ptr<TableIndexMeta> table_index_meta;
    std::string table_key;
    std::string index_key;
    status = GetTableIndexMeta(db_name, table_name, index_name, db_meta, table_meta, table_index_meta, &table_key, &index_key);
    if (!status.ok()) {
        return status;
    }

    SegmentIndexMeta segment_index_meta(segment_id, *table_index_meta);
    std::shared_ptr<MemIndex> mem_index = segment_index_meta.GetMemIndex();

    // Return when there is no mem index to dump.
    if (mem_index == nullptr || mem_index->IsDumping() || (mem_index->GetBaseMemIndex() == nullptr && mem_index->GetEMVBIndex() == nullptr) ||
        (begin_row_id != RowID() && mem_index->GetBaseMemIndex() != nullptr && begin_row_id != mem_index->GetBaseMemIndex()->GetBeginRowID())) {
        LOG_WARN(fmt::format("NewTxn::DumpMemIndex skipped dumping MemIndex {}.{}.{}.{}.{} since it doesn't exist or it is dumped (is_dumping: {})",
                             db_name,
                             table_name,
                             index_name,
                             segment_id,
                             begin_row_id.ToUint64(),
                             mem_index->IsDumping()));
        return Status::OK();
    }
    mem_index->SetIsDumping(true);

    ChunkIndexMetaInfo chunk_index_meta_info;
    if (mem_index->GetBaseMemIndex() != nullptr) {
        chunk_index_meta_info = mem_index->GetBaseMemIndex()->GetChunkIndexMetaInfo();
    } else if (mem_index->GetEMVBIndex() != nullptr) {
        chunk_index_meta_info = mem_index->GetEMVBIndex()->GetChunkIndexMetaInfo();
    } else {
        return Status::UnexpectedError("Invalid mem index.");
    }

    ChunkID chunk_id = 0;
    std::tie(chunk_id, status) = segment_index_meta.GetAndSetNextChunkID();
    if (!status.ok()) {
        return status;
    }

    std::vector<WalChunkIndexInfo> chunk_infos;
    chunk_infos.emplace_back(chunk_index_meta_info, chunk_id);

    // Dump Mem Index
    status = this->DumpSegmentMemIndex(segment_index_meta, chunk_id);
    if (!status.ok() && status.code() != ErrorCode::kEmptyMemIndex) {
        return status;
    }

    // Put the data into local txn store
    if (base_txn_store_ == nullptr) {
        base_txn_store_ = std::make_shared<DumpMemIndexTxnStore>();
        DumpMemIndexTxnStore *txn_store = static_cast<DumpMemIndexTxnStore *>(base_txn_store_.get());
        txn_store->db_name_ = db_name;
        txn_store->db_id_str_ = segment_index_meta.table_index_meta().table_meta().db_id_str();
        txn_store->table_name_ = table_name;
        txn_store->table_id_str_ = segment_index_meta.table_index_meta().table_meta().table_id_str();
        txn_store->index_name_ = index_name;
        txn_store->index_id_str_ = segment_index_meta.table_index_meta().index_id_str();
        txn_store->index_id_ = std::stoull(txn_store->index_id_str_);
        txn_store->table_key_ = table_key;
        txn_store->segment_ids_ = {segment_id};
        txn_store->chunk_infos_in_segments_.emplace(segment_id, chunk_infos);
    } else {
        DumpMemIndexTxnStore *txn_store = static_cast<DumpMemIndexTxnStore *>(base_txn_store_.get());
        txn_store->segment_ids_.emplace_back(segment_id);
        txn_store->chunk_infos_in_segments_.emplace(segment_id, chunk_infos);
    }

    return Status::OK();
}

Status NewTxn::OptimizeAllIndexes() {
    // TxnTimeStamp begin_ts = txn_context_ptr_->begin_ts_;
    CatalogMeta catalog_meta(this);
    std::vector<std::string> *db_id_strs_ptr = nullptr;
    std::vector<std::string> *db_names_ptr = nullptr;
    Status status = catalog_meta.GetDBIDs(db_id_strs_ptr, &db_names_ptr);
    if (!status.ok()) {
        return status;
    }
    for (size_t i = 0; i < db_id_strs_ptr->size(); ++i) {
        const std::string &db_id_str = (*db_id_strs_ptr)[i];
        const std::string &db_name = (*db_names_ptr)[i];

        DBMeta db_meta(db_id_str, db_name, this);
        std::vector<std::string> *table_id_strs_ptr = nullptr;
        std::vector<std::string> *table_names_ptr = nullptr;
        status = db_meta.GetTableIDs(table_id_strs_ptr, &table_names_ptr);
        if (!status.ok()) {
            return status;
        }
        for (size_t idx = 0; idx < table_id_strs_ptr->size(); ++idx) {
            // const std::string &table_id_str = (*table_id_strs_ptr)[idx];
            const std::string &table_name = (*table_names_ptr)[idx];

            status = this->OptimizeTableIndexes(db_name, table_name);
            if (!status.ok()) {
                return status;
            }
        }
    }

    return Status::OK();
}

Status NewTxn::OptimizeTableIndexes(const std::string &db_name, const std::string &table_name) {

    LOG_TRACE(fmt::format("Attempt to optimize the indexes of table: {}.{}", db_name, table_name));

    std::shared_ptr<DBMeta> db_meta;
    std::shared_ptr<TableMeta> table_meta;
    std::string table_key;
    TxnTimeStamp create_timestamp;
    Status status = GetTableMeta(db_name, table_name, db_meta, table_meta, create_timestamp, &table_key);
    if (!status.ok()) {
        return status;
    }
    std::vector<std::string> *index_id_strs_ptr = nullptr;
    std::vector<std::string> *index_names_ptr = nullptr;
    status = table_meta->GetIndexIDs(index_id_strs_ptr, &index_names_ptr);
    if (!status.ok()) {
        return status;
    }

    for (size_t i = 0; i < index_id_strs_ptr->size(); ++i) {
        const std::string &index_id_str = (*index_id_strs_ptr)[i];
        const std::string &index_name = (*index_names_ptr)[i];
        TableIndexMeta table_index_meta(index_id_str, index_name, *table_meta);

        std::vector<SegmentID> *segment_ids_ptr = nullptr;
        std::tie(segment_ids_ptr, status) = table_index_meta.GetSegmentIndexIDs1();
        if (!status.ok()) {
            return status;
        }
        for (SegmentID segment_id : *segment_ids_ptr) {
            SegmentIndexMeta segment_index_meta(segment_id, table_index_meta);
            status = this->OptimizeIndexInner(segment_index_meta, index_name, table_name, db_name, table_key);
            if (!status.ok()) {
                return status;
            }
        }
    }

    return Status::OK();
}

Status NewTxn::OptimizeIndex(const std::string &db_name, const std::string &table_name, const std::string &index_name, SegmentID segment_id) {
    Status status = Status::OK();

    std::shared_ptr<DBMeta> db_meta;
    std::shared_ptr<TableMeta> table_meta_opt;
    std::shared_ptr<TableIndexMeta> table_index_meta_opt;
    std::string table_key;
    std::string index_key;
    status = GetTableIndexMeta(db_name, table_name, index_name, db_meta, table_meta_opt, table_index_meta_opt, &table_key, &index_key);
    if (!status.ok()) {
        return status;
    }

    SegmentIndexMeta segment_index_meta(segment_id, *table_index_meta_opt);

    return OptimizeIndexInner(segment_index_meta, index_name, table_name, db_name, table_key);
}

Status NewTxn::OptimizeIndexInner(SegmentIndexMeta &segment_index_meta,
                                  const std::string &index_name,
                                  const std::string &table_name,
                                  const std::string &db_name,
                                  const std::string &table_key) {
    TableIndexMeta &table_index_meta = segment_index_meta.table_index_meta();
    TableMeta &table_meta = table_index_meta.table_meta();
    SegmentID segment_id = segment_index_meta.segment_id();

    auto [old_chunk_ids_ptr, status] = segment_index_meta.GetChunkIDs1();
    if (!status.ok()) {
        return status;
    }
    if (old_chunk_ids_ptr->size() <= 1) {
        return Status::OK();
    }
    RowID base_rowid;
    u32 row_cnt = 0;
    u32 term_cnt = 0;
    std::vector<size_t> row_cnts;

    std::string base_name;

    auto [index_base, index_status] = table_index_meta.GetIndexBase();
    if (!index_status.ok()) {
        return index_status;
    }
    if (index_base->index_type_ == IndexType::kFullText) {
        status = this->OptimizeFtIndex(index_base, segment_index_meta, base_rowid, row_cnt, term_cnt, base_name);
        if (!status.ok()) {
            return status;
        }
    } else {
        base_rowid = RowID(segment_id, 0);
        RowID last_rowid = base_rowid;
        ChunkIndexMetaInfo *chunk_info_ptr = nullptr;
        for (ChunkID old_chunk_id : *old_chunk_ids_ptr) {
            ChunkIndexMeta old_chunk_meta(old_chunk_id, segment_index_meta);
            {
                status = old_chunk_meta.GetChunkInfo(chunk_info_ptr);
                if (!status.ok()) {
                    return status;
                }
            }
            if (last_rowid != chunk_info_ptr->base_row_id_) {
                UnrecoverableError("OptimizeIndex: base_row_id is not continuous");
            }
            last_rowid.segment_offset_ += chunk_info_ptr->row_cnt_;
            row_cnt += chunk_info_ptr->row_cnt_;
            row_cnts.push_back(chunk_info_ptr->row_cnt_);
        }
    }
    std::vector<ChunkID> deprecate_ids = *old_chunk_ids_ptr;
    ChunkID chunk_id = 0;
    std::tie(chunk_id, status) = segment_index_meta.GetAndSetNextChunkID();
    if (!status.ok()) {
        return status;
    }

    std::optional<ChunkIndexMeta> chunk_index_meta;
    BufferObj *buffer_obj = nullptr;
    {
        status = NewCatalog::AddNewChunkIndex1(segment_index_meta,
                                               this,
                                               chunk_id,
                                               base_rowid,
                                               row_cnt,
                                               term_cnt,
                                               base_name,
                                               0 /*index_size*/,
                                               chunk_index_meta);
        if (!status.ok()) {
            return status;
        }
        status = chunk_index_meta->GetIndexBuffer(buffer_obj);
        if (!status.ok()) {
            return status;
        }
    }

    switch (index_base->index_type_) {
        case IndexType::kSecondary: {
            std::vector<std::pair<u32, BufferObj *>> old_buffers;
            for (size_t i = 0; i < deprecate_ids.size(); ++i) {
                ChunkID old_chunk_id = deprecate_ids[i];
                ChunkIndexMeta old_chunk_meta(old_chunk_id, segment_index_meta);

                BufferObj *buffer_obj = nullptr;
                {
                    // Status status = NewCatalog::GetChunkIndex(old_chunk_meta, buffer_obj);
                    status = old_chunk_meta.GetIndexBuffer(buffer_obj);
                    if (!status.ok()) {
                        return status;
                    }
                }
                old_buffers.emplace_back(row_cnts[i], buffer_obj);
            }

            BufferHandle buffer_handle = buffer_obj->Load();
            auto *data_ptr = static_cast<SecondaryIndexData *>(buffer_handle.GetDataMut());
            data_ptr->InsertMergeData(old_buffers);
            break;
        }
        case IndexType::kFullText: {
            // Skip here. merge finished in `OptimizeFtIndex`
            break;
        }
        case IndexType::kIVF: {
            SegmentMeta segment_meta(segment_id, table_meta);
            std::shared_ptr<ColumnDef> column_def;
            {
                auto [col_def, status] = table_index_meta.GetColumnDef();
                if (!status.ok()) {
                    return status;
                }
                column_def = std::move(col_def);
            }

            BufferHandle buffer_handle = buffer_obj->Load();
            auto *data_ptr = static_cast<IVFIndexInChunk *>(buffer_handle.GetDataMut());
            data_ptr->BuildIVFIndex(segment_meta, row_cnt, column_def);
            break;
        }
        case IndexType::kHnsw:
        case IndexType::kBMP: {
            SegmentMeta segment_meta(segment_id, table_meta);
            std::shared_ptr<ColumnDef> column_def;
            {
                auto [col_def, status] = table_index_meta.GetColumnDef();
                if (!status.ok()) {
                    return status;
                }
                column_def = std::move(col_def);
            }
            status = OptimizeVecIndex(index_base, column_def, segment_meta, base_rowid, row_cnt, buffer_obj);
            if (!status.ok()) {
                return status;
            }
            break;
        }
        case IndexType::kEMVB: {
            SegmentMeta segment_meta(segment_id, table_meta);
            std::shared_ptr<ColumnDef> column_def;
            {
                auto [col_def, status] = table_index_meta.GetColumnDef();
                if (!status.ok()) {
                    return status;
                }
                column_def = std::move(col_def);
            }

            BufferHandle buffer_handle = buffer_obj->Load();
            auto *data_ptr = static_cast<EMVBIndex *>(buffer_handle.GetDataMut());
            data_ptr->BuildEMVBIndex(base_rowid, row_cnt, segment_meta, column_def);
            break;
        }
        default: {
            UnrecoverableError("Not implemented yet");
        }
    }
    {
        // To delete deprecated chunk ids
        status = segment_index_meta.RemoveChunkIDs(deprecate_ids);
        if (!status.ok()) {
            return status;
        }
    }

    buffer_obj->Save();
    if (index_base->index_type_ == IndexType::kHnsw || index_base->index_type_ == IndexType::kBMP) {
        if (buffer_obj->type() != BufferType::kMmap) {
            buffer_obj->ToMmap();
        }
    }

    std::vector<WalChunkIndexInfo> chunk_infos;
    chunk_infos.emplace_back(*chunk_index_meta);

    // Put the data into local txn store
    if (base_txn_store_ == nullptr) {
        base_txn_store_ = std::make_shared<OptimizeIndexTxnStore>();
    }

    OptimizeIndexTxnStore *optimize_index_txn_store = static_cast<OptimizeIndexTxnStore *>(base_txn_store_.get());
    if (std::find(optimize_index_txn_store->db_names_.begin(), optimize_index_txn_store->db_names_.end(), db_name) ==
        optimize_index_txn_store->db_names_.end()) {
        optimize_index_txn_store->db_names_.emplace_back(db_name);
    }
    if (std::find(optimize_index_txn_store->table_names_in_db_[db_name].begin(),
                  optimize_index_txn_store->table_names_in_db_[db_name].end(),
                  table_name) == optimize_index_txn_store->table_names_in_db_[db_name].end()) {
        optimize_index_txn_store->table_names_in_db_[db_name].emplace_back(table_name);
    }

    optimize_index_txn_store->entries_.emplace_back(db_name,
                                                    table_meta.db_id_str(),
                                                    std::stoull(table_meta.db_id_str()),
                                                    table_name,
                                                    table_meta.table_id_str(),
                                                    std::stoull(table_meta.table_id_str()),
                                                    table_key,
                                                    index_name,
                                                    table_index_meta.index_id_str(),
                                                    std::stoull(table_index_meta.index_id_str()),
                                                    segment_id,
                                                    std::move(chunk_infos),
                                                    std::move(deprecate_ids));

    return Status::OK();
}

Status NewTxn::AlterIndexByParams(const std::string &db_name,
                                  const std::string &table_name,
                                  const std::string &index_name,
                                  std::vector<std::unique_ptr<InitParameter>> raw_params) {
    Status status = Status::OK();

    std::shared_ptr<DBMeta> db_meta;
    std::shared_ptr<TableMeta> table_meta_opt;
    std::shared_ptr<TableIndexMeta> table_index_meta_opt;
    std::string table_key;
    std::string index_key;
    status = GetTableIndexMeta(db_name, table_name, index_name, db_meta, table_meta_opt, table_index_meta_opt, &table_key, &index_key);
    if (!status.ok()) {
        return status;
    }
    TableIndexMeta &table_index_meta = *table_index_meta_opt;

    auto [index_base, index_status] = table_index_meta.GetIndexBase();
    if (!index_status.ok()) {
        return index_status;
    }

    bool opt = false;
    std::shared_ptr<IndexBase> new_index_base;
    switch (index_base->index_type_) {
        case IndexType::kBMP: {
            opt = true;
            break;
        }
        case IndexType::kHnsw: {
            auto params = HnswUtil::ParseOptimizeOptions(raw_params);
            if (!params) {
                break;
            }
            opt = true;
            if (params->compress_to_lvq) {
                auto *hnsw_index = static_cast<IndexHnsw *>(index_base.get());
                if (hnsw_index->encode_type_ != HnswEncodeType::kPlain) {
                    LOG_WARN("Not implemented");
                    break;
                }
                auto new_index_hnsw = std::make_shared<IndexHnsw>(*hnsw_index);
                // IndexHnsw old_index_hnsw = *hnsw_index;
                new_index_hnsw->encode_type_ = HnswEncodeType::kLVQ;
                if (new_index_hnsw->build_type_ == HnswBuildType::kLSG) {
                    new_index_hnsw->build_type_ = HnswBuildType::kPlain;
                }
                new_index_base = std::move(new_index_hnsw);
            } else if (params->compress_to_rabitq) {
                auto *hnsw_index = static_cast<IndexHnsw *>(index_base.get());
                if (hnsw_index->encode_type_ != HnswEncodeType::kPlain) {
                    LOG_WARN("Not implemented");
                    break;
                }
                auto new_index_hnsw = std::make_shared<IndexHnsw>(*hnsw_index);
                // IndexHnsw old_index_hnsw = *hnsw_index;
                new_index_hnsw->encode_type_ = HnswEncodeType::kRabitq;
                if (new_index_hnsw->build_type_ == HnswBuildType::kLSG) {
                    new_index_hnsw->build_type_ = HnswBuildType::kPlain;
                }
                new_index_base = std::move(new_index_hnsw);
            }
            break;
        }
        default: {
        }
    }
    if (!opt) {
        return Status::OK();
    }
    std::vector<SegmentID> *segment_ids_ptr = nullptr;
    std::tie(segment_ids_ptr, status) = table_index_meta.GetSegmentIndexIDs1();
    if (!status.ok()) {
        return status;
    }

    for (SegmentID segment_id : *segment_ids_ptr) {
        SegmentIndexMeta segment_index_meta(segment_id, table_index_meta);
        Status status = AlterSegmentIndexByParams(segment_index_meta, raw_params);
        if (!status.ok()) {
            return status;
        }
    }

    if (new_index_base) {
        Status status = table_index_meta.SetIndexBase(new_index_base);
        if (!status.ok()) {
            return status;
        }
    }

    // Put the data into local txn store
    if (base_txn_store_ == nullptr) {
        base_txn_store_ = std::make_shared<AlterIndexTxnStore>();
    }

    AlterIndexTxnStore *alter_index_txn_store = static_cast<AlterIndexTxnStore *>(base_txn_store_.get());
    alter_index_txn_store->db_name_ = db_name;
    alter_index_txn_store->db_id_str_ = db_meta->db_id_str();
    alter_index_txn_store->db_id_ = std::stoull(db_meta->db_id_str());
    alter_index_txn_store->table_name_ = table_name;
    alter_index_txn_store->table_id_str_ = table_meta_opt->table_id_str();
    alter_index_txn_store->table_id_ = std::stoull(table_meta_opt->table_id_str());
    alter_index_txn_store->index_name_ = index_name;
    alter_index_txn_store->index_id_str_ = table_index_meta_opt->index_id_str();
    alter_index_txn_store->index_id_ = std::stoull(table_index_meta_opt->index_id_str());
    alter_index_txn_store->params_ = std::move(raw_params);

    return Status::OK();
}

Status NewTxn::ListIndex(const std::string &db_name, const std::string &table_name, std::vector<std::string> &index_names) {
    std::shared_ptr<DBMeta> db_meta;
    std::shared_ptr<TableMeta> table_meta;
    TxnTimeStamp create_timestamp;
    Status status = GetTableMeta(db_name, table_name, db_meta, table_meta, create_timestamp);
    if (!status.ok()) {
        return status;
    }
    std::vector<std::string> *index_id_strs_ptr = nullptr;
    std::vector<std::string> *index_names_ptr = nullptr;
    status = table_meta->GetIndexIDs(index_id_strs_ptr, &index_names_ptr);
    if (!status.ok()) {
        return status;
    }
    index_names = *index_names_ptr;
    return Status::OK();
}

Status NewTxn::AppendIndex(TableIndexMeta &table_index_meta, const std::pair<RowID, u64> &append_range) {
    auto [index_base, index_status] = table_index_meta.GetIndexBase();
    if (!index_status.ok()) {
        return index_status;
    }
    size_t column_idx = 0;
    {
        auto [column_def, status] = table_index_meta.table_meta().GetColumnDefByColumnName(index_base->column_name(), &column_idx);
        if (!status.ok()) {
            return status;
        }
    }
    std::optional<SegmentMeta> segment_meta;
    std::optional<SegmentIndexMeta> segment_index_meta;
    std::optional<BlockMeta> block_meta;
    std::optional<ColumnMeta> column_meta;
    BlockOffset cur_offset = 0;
    BlockOffset cur_row_cnt = 0;

    auto append_in_column = [&]() {
        ColumnVector col;
        {
            Status status =
                NewCatalog::GetColumnVector(*column_meta, column_meta->get_column_def(), cur_offset + cur_row_cnt, ColumnVectorMode::kReadOnly, col);
            if (!status.ok()) {
                return status;
            }
        }
        BlockID block_id = block_meta->block_id();
        {
            Status status = this->AppendMemIndex(*segment_index_meta, block_id, col, cur_offset, cur_row_cnt);
            if (!status.ok()) {
                return status;
            }
        }
        return Status::OK();
    };

    SegmentID segment_id = append_range.first.segment_id_;
    BlockID block_id = append_range.first.segment_offset_ >> BLOCK_OFFSET_SHIFT;
    cur_offset = append_range.first.segment_offset_ & BLOCK_OFFSET_MASK;
    cur_row_cnt = append_range.second;
    if (!segment_meta || segment_meta->segment_id() != segment_id) {
        segment_meta.emplace(segment_id, table_index_meta.table_meta());
        if (!table_index_meta.HasSegmentIndexID(segment_id) && append_range.first.segment_offset_ == 0) {
            Status status = NewCatalog::AddNewSegmentIndex1(table_index_meta, this, segment_id, segment_index_meta);
            if (!status.ok()) {
                return status;
            }
        } else {
            segment_index_meta.emplace(segment_id, table_index_meta);
        }
    }
    if (!block_meta || block_meta->block_id() != block_id) {
        block_meta.emplace(block_id, segment_meta.value());
        column_meta.emplace(column_idx, block_meta.value());
    }
    Status status = append_in_column();
    if (!status.ok()) {
        return status;
    }

    return Status::OK();
}

Status
NewTxn::AppendMemIndex(SegmentIndexMeta &segment_index_meta, BlockID block_id, const ColumnVector &col, BlockOffset offset, BlockOffset row_cnt) {
    SegmentOffset block_offset = block_id * DEFAULT_BLOCK_CAPACITY;
    RowID base_row_id = RowID(segment_index_meta.segment_id(), block_offset + offset);

    TxnTimeStamp begin_ts = txn_context_ptr_->begin_ts_;
    auto [index_base, index_status] = segment_index_meta.table_index_meta().GetIndexBase();
    if (!index_status.ok()) {
        return index_status;
    }
    std::shared_ptr<MemIndex> mem_index = segment_index_meta.GetMemIndex(true);
    bool is_null = mem_index->IsNull();
    LOG_TRACE(fmt::format("NewTxn::AppendMemIndex UpdateBegin mem_index {:p}, is_null {}", (void *)mem_index.get(), is_null));
    switch (index_base->index_type_) {
        case IndexType::kSecondary: {
            std::shared_ptr<SecondaryIndexInMem> memory_secondary_index;
            if (is_null) {
                auto [column_def, status] = segment_index_meta.table_index_meta().GetColumnDef();
                if (!status.ok()) {
                    return status;
                }
                memory_secondary_index = SecondaryIndexInMem::NewSecondaryIndexInMem(column_def, base_row_id);

                mem_index->SetSecondaryIndex(memory_secondary_index);
            } else {
                memory_secondary_index = mem_index->GetSecondaryIndex();
            }
            memory_secondary_index->InsertBlockData(block_offset, col, offset, row_cnt);
            break;
        }
        case IndexType::kFullText: {
            const auto *index_fulltext = static_cast<const IndexFullText *>(index_base.get());
            std::shared_ptr<MemoryIndexer> memory_indexer;
            bool need_to_update_ft_segment_ts = false;
            if (is_null) {
                auto [column_def, status] = segment_index_meta.table_index_meta().GetColumnDef();
                if (!status.ok()) {
                    return status;
                }

                std::shared_ptr<std::string> index_dir = segment_index_meta.GetSegmentIndexDir();
                std::string base_name = fmt::format("ft_{:016x}", base_row_id.ToUint64());
                std::string full_path = fmt::format("{}/{}", InfinityContext::instance().config()->DataDir(), *index_dir);
                memory_indexer = std::make_unique<MemoryIndexer>(full_path, base_name, base_row_id, index_fulltext->flag_, index_fulltext->analyzer_);
                need_to_update_ft_segment_ts = true;
                mem_index->SetFulltextIndex(memory_indexer);
            } else {
                memory_indexer = mem_index->GetFulltextIndex();
                LOG_TRACE(fmt::format("AppendMemIndex: memory_indexer_ is not null, base_row_id: {}, doc_count: {}",
                                      base_row_id.ToUint64(),
                                      memory_indexer->GetDocCount()));
                RowID exp_begin_row_id = memory_indexer->GetBeginRowID() + memory_indexer->GetDocCount();
                assert(base_row_id >= exp_begin_row_id);
                if (base_row_id > exp_begin_row_id) {
                    LOG_WARN(fmt::format("Begin row id: {}, expect begin row id: {}, insert gap: {}",
                                         base_row_id.ToUint64(),
                                         exp_begin_row_id.ToUint64(),
                                         base_row_id - exp_begin_row_id));
                    memory_indexer->InsertGap(base_row_id - exp_begin_row_id);
                }
            }

            auto col_ptr = std::make_shared<ColumnVector>(std::move(col));
            if (index_fulltext->IsRealtime()) {
                std::unique_ptr<std::binary_semaphore> sema = memory_indexer->AsyncInsert(col_ptr, offset, row_cnt);
                this->AddSemaphore(std::move(sema));
            } else {
                // mem_index->GetFulltextIndex()->Insert(col_ptr, offset, row_cnt, false);
                std::shared_ptr<AppendMemIndexTask> append_mem_index_task = std::make_shared<AppendMemIndexTask>(mem_index, col_ptr, offset, row_cnt);
                memory_indexer->AsyncInsertTop(append_mem_index_task.get());
                auto *mem_index_appender = InfinityContext::instance().storage()->mem_index_appender();
                mem_index_appender->Submit(append_mem_index_task);
            }
            if (need_to_update_ft_segment_ts) {
                // To avoid deadlock of mem index mutex and table index reader cache mutex, update the ts here.
                // query will lock table index reader cache mutex first, then mem index mutex.
                // append will lock mem index mutex first, then table index reader cache mutex.
                segment_index_meta.table_index_meta().table_meta().InvalidateFtIndexCache();
            }
            break;
        }
        case IndexType::kIVF: {
            std::shared_ptr<IVFIndexInMem> memory_ivf_index;
            if (is_null) {
                auto [column_def, status] = segment_index_meta.table_index_meta().GetColumnDef();
                if (!status.ok()) {
                    return status;
                }
                memory_ivf_index = IVFIndexInMem::NewIVFIndexInMem(column_def.get(), index_base.get(), base_row_id);
                mem_index->SetIVFIndex(memory_ivf_index);

            } else {
                memory_ivf_index = mem_index->GetIVFIndex();
            }
            memory_ivf_index->InsertBlockData(block_offset, col, offset, row_cnt);
            break;
        }
        case IndexType::kHnsw: {
            std::shared_ptr<HnswIndexInMem> memory_hnsw_index;
            if (is_null) {
                auto [column_def, status] = segment_index_meta.table_index_meta().GetColumnDef();
                if (!status.ok()) {
                    return status;
                }
                memory_hnsw_index = HnswIndexInMem::Make(base_row_id, index_base.get(), column_def);
                mem_index->SetHnswIndex(memory_hnsw_index);
            } else {
                memory_hnsw_index = mem_index->GetHnswIndex();
            }
            memory_hnsw_index->InsertVecs(block_offset, col, offset, row_cnt);
            break;
        }
        case IndexType::kBMP: {
            std::shared_ptr<BMPIndexInMem> memory_bmp_index;
            if (is_null) {
                auto [column_def, status] = segment_index_meta.table_index_meta().GetColumnDef();
                if (!status.ok()) {
                    return status;
                }
                memory_bmp_index = std::make_shared<BMPIndexInMem>(base_row_id, index_base.get(), column_def.get());
                mem_index->SetBMPIndex(memory_bmp_index);
            } else {
                memory_bmp_index = mem_index->GetBMPIndex();
            }
            memory_bmp_index->AddDocs(block_offset, col, offset, row_cnt);
            break;
        }
        case IndexType::kEMVB: {
            std::shared_ptr<EMVBIndexInMem> memory_emvb_index;
            TableMeta &table_meta = segment_index_meta.table_index_meta().table_meta();
            if (is_null) {
                auto [column_def, status] = segment_index_meta.table_index_meta().GetColumnDef();
                if (!status.ok()) {
                    return status;
                }
                memory_emvb_index = EMVBIndexInMem::NewEMVBIndexInMem(index_base, column_def, base_row_id);
                memory_emvb_index->SetSegmentID(table_meta.db_id_str(), table_meta.table_id_str(), segment_index_meta.segment_id());
                mem_index->SetEMVBIndex(memory_emvb_index);
            } else {
                memory_emvb_index = mem_index->GetEMVBIndex();
            }
            MetaCache *meta_cache = table_meta.meta_cache();
            memory_emvb_index->Insert(col, offset, row_cnt, segment_index_meta.kv_instance(), begin_ts, meta_cache);
            break;
        }
        default: {
            UnrecoverableError("Not implemented yet");
        }
    }
    mem_index->UpdateEnd();
    LOG_TRACE(fmt::format("NewTxn::AppendMemIndex UpdateEnd mem_index {:p}", (void *)mem_index.get()));

    // Trigger dump if necessary
    if (!this->IsReplay()) {
        size_t row_count = mem_index->GetRowCount();
        size_t row_quota = InfinityContext::instance().config()->MemIndexCapacity();
        if (row_count >= row_quota) {
            TableMeta &table_meta = segment_index_meta.table_index_meta().table_meta();
            auto [db_name, table_name] = table_meta.GetDBTableName();
            auto [index_base, _] = segment_index_meta.table_index_meta().GetIndexBase();
            std::string index_name = *index_base->index_name_;
            SegmentID segment_id = segment_index_meta.segment_id();
            RowID begin_row_id = mem_index->GetBeginRowID();
            std::shared_ptr<DumpMemIndexTask> dump_task =
                std::make_shared<DumpMemIndexTask>(db_name, table_name, index_name, segment_id, begin_row_id);
            DumpIndexProcessor *dump_index_processor = InfinityContext::instance().storage()->dump_index_processor();
            LOG_INFO(fmt::format("MemIndex row count {} exceeds quota {}.  Submit dump task: {}", row_count, row_quota, dump_task->ToString()));
            dump_index_processor->Submit(std::move(dump_task));
        }
    }

    return Status::OK();
}

Status NewTxn::PopulateIndex(const std::string &db_name,
                             const std::string &table_name,
                             const std::string &index_name,
                             const std::string &table_key,
                             TableIndexMeta &table_index_meta,
                             SegmentMeta &segment_meta,
                             size_t segment_row_cnt,
                             DumpIndexCause dump_index_cause,
                             WalCmdCreateIndexV2 *create_index_cmd_ptr) {
    // PopulateIndex is used in create index / import and compact
    std::optional<SegmentIndexMeta> segment_index_meta;
    {
        auto status = NewCatalog::AddNewSegmentIndex1(table_index_meta, this, segment_meta.segment_id(), segment_index_meta);
        if (!status.ok()) {
            return status;
        }
    }
    auto [index_base, index_status] = table_index_meta.GetIndexBase();
    if (!index_status.ok()) {
        return index_status;
    }
    std::shared_ptr<ColumnDef> column_def;
    ColumnID column_id = 0;
    {
        auto [column_def_, status] = table_index_meta.table_meta().GetColumnDefByColumnName(index_base->column_name());
        if (!status.ok()) {
            return status;
        }
        column_def = column_def_;
        column_id = column_def->id();
    }
    std::vector<ChunkID> old_chunk_ids;
    {
        auto [old_chunk_ids_ptr, status] = segment_index_meta->GetChunkIDs1();
        if (!status.ok()) {
            return status;
        }
        old_chunk_ids = *old_chunk_ids_ptr;
    }
    std::vector<ChunkID> new_chunk_ids;
    if (index_base->index_type_ == IndexType::kIVF) {
        Status status = this->PopulateIvfIndexInner(index_base, *segment_index_meta, segment_meta, column_def, new_chunk_ids);
        if (!status.ok()) {
            return status;
        }
    } else if (index_base->index_type_ == IndexType::kEMVB) {
        Status status = this->PopulateEmvbIndexInner(index_base, *segment_index_meta, segment_meta, column_def, new_chunk_ids);
        if (!status.ok()) {
            return status;
        }
    } else {
        switch (index_base->index_type_) {
            case IndexType::kSecondary:
            case IndexType::kHnsw:
            case IndexType::kBMP: {
                Status status = this->PopulateIndexToMem(*segment_index_meta, segment_meta, column_id, segment_row_cnt);
                if (!status.ok()) {
                    return status;
                }
                break;
            }
            case IndexType::kFullText: {
                Status status = this->PopulateFtIndexInner(index_base, *segment_index_meta, segment_meta, column_id, segment_row_cnt, new_chunk_ids);
                if (!status.ok()) {
                    return status;
                }
                break;
            }
            case IndexType::kDiskAnn: { // TODO
                LOG_WARN("Not implemented yet");
                return Status::OK();
            }
            default: {
                UnrecoverableError("Invalid index type");
                return Status::OK();
            }
        }
        {
            Status status;
            ChunkID new_chunk_id = 0;
            std::tie(new_chunk_id, status) = segment_index_meta->GetAndSetNextChunkID();
            if (!status.ok()) {
                return status;
            }

            status = DumpSegmentMemIndex(*segment_index_meta, new_chunk_id);
            if (!status.ok()) {
                if (status.code() != ErrorCode::kEmptyMemIndex) {
                    return status;
                }
            } else {
                new_chunk_ids.push_back(new_chunk_id);
            }
        }
    }

    std::vector<WalChunkIndexInfo> chunk_infos;
    for (auto new_chunk_id : new_chunk_ids) {
        ChunkIndexMeta chunk_index_meta(new_chunk_id, *segment_index_meta);
        chunk_infos.emplace_back(chunk_index_meta);
    }

    // Put the index info local txn store
    switch (dump_index_cause) {
        case DumpIndexCause::kCompact: {
            CompactTxnStore *compact_txn_store = static_cast<CompactTxnStore *>(base_txn_store_.get());
            compact_txn_store->chunk_infos_in_segments_.emplace(segment_meta.segment_id(), chunk_infos);
            compact_txn_store->deprecate_ids_in_segments_.emplace(segment_meta.segment_id(), old_chunk_ids);
            break;
        }
        case DumpIndexCause::kImport: {
            ImportTxnStore *import_txn_store = static_cast<ImportTxnStore *>(base_txn_store_.get());
            import_txn_store->chunk_infos_in_segments_.emplace(segment_meta.segment_id(), chunk_infos);
            import_txn_store->deprecate_ids_in_segments_.emplace(segment_meta.segment_id(), old_chunk_ids);
        }
        default: {
        }
    }

    if (create_index_cmd_ptr) {
        WalCmdCreateIndexV2 &create_index_cmd = *create_index_cmd_ptr;
        create_index_cmd.segment_index_infos_.emplace_back(segment_meta.segment_id(), std::move(chunk_infos));
    }

    return Status::OK();
}

Status NewTxn::ReplayDumpIndex(WalCmdDumpIndexV2 *dump_index_cmd) {
    Status status;
    std::shared_ptr<DBMeta> db_meta;
    std::shared_ptr<TableMeta> table_meta;
    std::shared_ptr<TableIndexMeta> table_index_meta;
    std::string table_key;
    std::string index_key;
    status = GetTableIndexMeta(dump_index_cmd->db_name_,
                               dump_index_cmd->table_name_,
                               dump_index_cmd->index_name_,
                               db_meta,
                               table_meta,
                               table_index_meta,
                               &table_key,
                               &index_key);
    if (!status.ok()) {
        return status;
    }

    SegmentID segment_id = dump_index_cmd->segment_id_;
    std::optional<SegmentIndexMeta> segment_index_meta_opt;
    {
        auto [segment_ids_ptr, status] = table_index_meta->GetSegmentIndexIDs1();
        if (!status.ok()) {
            return status;
        }
        auto iter = std::find(segment_ids_ptr->begin(), segment_ids_ptr->end(), segment_id);
        if (iter == segment_ids_ptr->end()) {
            status = NewCatalog::AddNewSegmentIndex1(*table_index_meta, this, segment_id, segment_index_meta_opt);
            if (!status.ok()) {
                return status;
            }
        } else {
            segment_index_meta_opt.emplace(segment_id, *table_index_meta);
        }
    }
    SegmentIndexMeta &segment_index_meta = *segment_index_meta_opt;

    std::vector<ChunkID> chunk_ids_to_delete;
    std::vector<ChunkID> *chunk_ids_ptr = nullptr;
    {
        std::unordered_set<ChunkID> deprecate_chunk_ids(dump_index_cmd->deprecate_ids_.begin(), dump_index_cmd->deprecate_ids_.end());
        std::tie(chunk_ids_ptr, status) = segment_index_meta.GetChunkIDs1();
        if (!status.ok()) {
            return status;
        }
        for (ChunkID chunk_id : *chunk_ids_ptr) {
            if (deprecate_chunk_ids.contains(chunk_id)) {
                chunk_ids_to_delete.push_back(chunk_id);
            }
        }
    }

    // Remove old ones;
    status = segment_index_meta.RemoveChunkIDs(chunk_ids_to_delete);
    if (!status.ok()) {
        return status;
    }

    ChunkID next_chunk_id = 0;
    ChunkID max_chunk_id = 0;
    status = segment_index_meta.GetNextChunkID(next_chunk_id);
    if (!status.ok() && status.code_ != ErrorCode::kNotFound) {
        return status;
    }

    for (const WalChunkIndexInfo &chunk_info : dump_index_cmd->chunk_infos_) {
        status = NewCatalog::LoadFlushedChunkIndex1(segment_index_meta, chunk_info, this);
        if (!status.ok()) {
            std::pair<RowID, u64> append_range = {chunk_info.base_rowid_, chunk_info.row_count_};
            status = this->AppendIndex(*table_index_meta, append_range);
            if (!status.ok()) {
                return status;
            }
            // Dump Mem Index
            status = this->DumpSegmentMemIndex(segment_index_meta, chunk_info.chunk_id_);
            if (!status.ok()) {
                return status;
            }
        }
        if (max_chunk_id < chunk_info.chunk_id_) {
            max_chunk_id = chunk_info.chunk_id_;
        }
    }
    if (next_chunk_id <= max_chunk_id) {
        status = segment_index_meta.SetNextChunkID(max_chunk_id + 1);
        if (!status.ok()) {
            return status;
        }
    }

    std::shared_ptr<MemIndex> mem_index = segment_index_meta.PopMemIndex();
    if (mem_index != nullptr) {
        mem_index->ClearMemIndex();
    }
    return Status::OK();
}

Status NewTxn::InitSegmentIndex(SegmentIndexMeta &segment_index_meta, SegmentMeta &segment_meta) {
    Status status;
    std::shared_ptr<MemIndex> mem_index = segment_index_meta.GetMemIndex();
    std::shared_ptr<IndexBase> index_base;
    std::tie(index_base, status) = segment_index_meta.table_index_meta().GetIndexBase();
    if (!status.ok()) {
        return status;
    }

    if (index_base->index_type_ == IndexType::kHnsw) {
        const auto *index_hnsw = static_cast<const IndexHnsw *>(index_base.get());
        if (index_hnsw->build_type_ == HnswBuildType::kLSG) {
            size_t segment_row_cnt = 0;
            std::tie(segment_row_cnt, status) = segment_meta.GetRowCnt1();
            if (!status.ok()) {
                return status;
            }
            std::shared_ptr<ColumnDef> column_def;
            std::tie(column_def, status) = segment_index_meta.table_index_meta().GetColumnDef();
            if (!status.ok()) {
                return status;
            }
            ColumnID column_id = column_def->id();
            std::vector<BlockID> *block_ids_ptr = nullptr;
            std::tie(block_ids_ptr, status) = segment_meta.GetBlockIDs1();
            size_t block_capacity = DEFAULT_BLOCK_CAPACITY;

            // Get random vector for sampling
            {
                float sample_radio = index_hnsw->lsg_config_->sample_ratio_;
                size_t max_sample_num = segment_row_cnt * sample_radio;
                if (max_sample_num <= 0) {
                    return Status::OK();
                }
                for (BlockID block_id : *block_ids_ptr) {
                    if (max_sample_num <= 0) {
                        break;
                    }
                    size_t row_cnt =
                        block_id == block_ids_ptr->back() ? segment_row_cnt - block_capacity * (block_ids_ptr->size() - 1) : block_capacity;
                    BlockMeta block_meta(block_id, segment_meta);
                    ColumnMeta column_meta(column_id, block_meta);
                    ColumnVector col;
                    status = NewCatalog::GetColumnVector(column_meta, column_meta.get_column_def(), row_cnt, ColumnVectorMode::kReadOnly, col);
                    if (!status.ok()) {
                        return status;
                    }
                    BlockOffset offset = 0;
                    SegmentOffset block_offset = block_id * DEFAULT_BLOCK_CAPACITY;
                    RowID base_row_id = RowID(segment_index_meta.segment_id(), block_offset + offset);

                    // Make memory index
                    std::shared_ptr<HnswIndexInMem> memory_hnsw_index;
                    if (mem_index->IsNull()) {
                        memory_hnsw_index = HnswIndexInMem::Make(base_row_id, index_base.get(), column_def);
                        mem_index->SetHnswIndex(memory_hnsw_index);
                    } else {
                        memory_hnsw_index = mem_index->GetHnswIndex();
                    }

                    // Sampling
                    size_t block_sample_num = memory_hnsw_index->InsertSampleVecs(max_sample_num, block_offset, offset, col, row_cnt);
                    max_sample_num -= block_sample_num;
                }
            }

            // Get memory index
            std::shared_ptr<HnswIndexInMem> memory_hnsw_index;
            if (mem_index->IsNull()) {
                UnrecoverableError("memory index is null, not init in InsertSampleVecs");
            } else {
                memory_hnsw_index = mem_index->GetHnswIndex();
            }

            // Compute neighborhood radius by HnswLSGBuilder
            {
                for (BlockID block_id : *block_ids_ptr) {
                    size_t row_cnt =
                        block_id == block_ids_ptr->back() ? segment_row_cnt - block_capacity * (block_ids_ptr->size() - 1) : block_capacity;
                    BlockMeta block_meta(block_id, segment_meta);
                    ColumnMeta column_meta(column_id, block_meta);
                    ColumnVector col;
                    status = NewCatalog::GetColumnVector(column_meta, column_meta.get_column_def(), row_cnt, ColumnVectorMode::kReadOnly, col);
                    if (!status.ok()) {
                        return status;
                    }
                    BlockOffset offset = 0;
                    SegmentOffset block_offset = block_id * DEFAULT_BLOCK_CAPACITY;

                    // Sampling
                    memory_hnsw_index->InsertLSAvg(block_offset, offset, col, row_cnt);
                }
            }

            memory_hnsw_index->SetLSGParam();
        }
    }
    return Status::OK();
}

Status NewTxn::PopulateIndexToMem(SegmentIndexMeta &segment_index_meta, SegmentMeta &segment_meta, ColumnID column_id, size_t segment_row_cnt) {
    InitSegmentIndex(segment_index_meta, segment_meta);

    auto [block_ids, status] = segment_meta.GetBlockIDs1();
    if (!status.ok()) {
        return status;
    }
    size_t block_capacity = DEFAULT_BLOCK_CAPACITY;
    for (BlockID block_id : *block_ids) {
        BlockMeta block_meta(block_id, segment_meta);
        ColumnMeta column_meta(column_id, block_meta);

        size_t row_cnt = block_id == block_ids->back() ? segment_row_cnt - block_capacity * (block_ids->size() - 1) : block_capacity;

        ColumnVector col;
        status = NewCatalog::GetColumnVector(column_meta, column_meta.get_column_def(), row_cnt, ColumnVectorMode::kReadOnly, col);
        if (!status.ok()) {
            return status;
        }
        u32 offset = 0;
        status = this->AppendMemIndex(segment_index_meta, block_id, col, offset, row_cnt);
        if (!status.ok()) {
            return status;
        }
    }
    return Status::OK();
}

Status NewTxn::PopulateFtIndexInner(std::shared_ptr<IndexBase> index_base,
                                    SegmentIndexMeta &segment_index_meta,
                                    SegmentMeta &segment_meta,
                                    ColumnID column_id,
                                    size_t segment_row_cnt,
                                    std::vector<ChunkID> &new_chunk_ids) {
    if (index_base->index_type_ != IndexType::kFullText) {
        UnrecoverableError("Invalid index type");
    }
    const IndexFullText *index_fulltext = static_cast<const IndexFullText *>(index_base.get());
    Status status;
    std::shared_ptr<MemIndex> mem_index = segment_index_meta.GetMemIndex();
    std::vector<BlockID> *block_ids_ptr = nullptr;
    std::tie(block_ids_ptr, status) = segment_meta.GetBlockIDs1();
    if (!status.ok()) {
        return status;
    }
    size_t block_capacity = DEFAULT_BLOCK_CAPACITY;
    i64 mem_index_capacity = InfinityContext::instance().storage()->config()->MemIndexCapacity();
    std::shared_ptr<std::string> index_dir = segment_index_meta.GetSegmentIndexDir();
    std::shared_ptr<MemoryIndexer> memory_indexer = nullptr;
    for (BlockID block_id : *block_ids_ptr) {
        if (memory_indexer == nullptr) {
            RowID base_row_id(segment_index_meta.segment_id(), block_id * block_capacity);
            std::string base_name = fmt::format("ft_{:016x}", base_row_id.ToUint64());
            std::string full_path = fmt::format("{}/{}", InfinityContext::instance().config()->DataDir(), *index_dir);
            memory_indexer = std::make_shared<MemoryIndexer>(full_path, base_name, base_row_id, index_fulltext->flag_, index_fulltext->analyzer_);
            LOG_INFO(fmt::format("PopulateFtIndexInner created memory_indexer, base_name: {}", base_name));
        }
        BlockMeta block_meta(block_id, segment_meta);
        ColumnMeta column_meta(column_id, block_meta);

        size_t row_cnt = block_id == block_ids_ptr->back() ? segment_row_cnt - block_capacity * (block_ids_ptr->size() - 1) : block_capacity;

        ColumnVector col;
        status = NewCatalog::GetColumnVector(column_meta, column_meta.get_column_def(), row_cnt, ColumnVectorMode::kReadOnly, col);
        if (!status.ok()) {
            return status;
        }
        SegmentOffset block_offset = block_meta.block_capacity() * block_meta.block_id();
        RowID begin_row_id(segment_meta.segment_id(), block_offset);

        RowID exp_begin_row_id = memory_indexer->GetBeginRowID() + memory_indexer->GetDocCount();
        assert(begin_row_id >= exp_begin_row_id);
        if (begin_row_id > exp_begin_row_id) {
            LOG_WARN(fmt::format("Begin row id: {}, expect begin row id: {}, insert gap: {}",
                                 begin_row_id.ToUint64(),
                                 exp_begin_row_id.ToUint64(),
                                 begin_row_id - exp_begin_row_id));
            memory_indexer->InsertGap(begin_row_id - exp_begin_row_id);
        }

        auto col_ptr = std::make_shared<ColumnVector>(std::move(col));
        memory_indexer->Insert(col_ptr, 0, row_cnt, false /*offline*/);
        memory_indexer->Commit(false /*offline*/);
        if (memory_indexer->GetRowCount() >= size_t(mem_index_capacity)) {
            LOG_INFO(fmt::format("PopulateFtIndexInner dump memory_indexer, base_name: {}, current row count: {}",
                                 memory_indexer->GetBaseName(),
                                 memory_indexer->GetRowCount()));
            memory_indexer->Dump(false /*offline*/, false /*spill*/);
            ChunkID new_chunk_id = 0;
            std::tie(new_chunk_id, status) = segment_index_meta.GetAndSetNextChunkID();
            if (!status.ok()) {
                return status;
            }
            new_chunk_ids.push_back(new_chunk_id);

            std::optional<ChunkIndexMeta> chunk_index_meta;
            ChunkIndexMetaInfo chunk_index_meta_info = memory_indexer->GetChunkIndexMetaInfo();
            status = NewCatalog::AddNewChunkIndex1(segment_index_meta,
                                                   this,
                                                   new_chunk_id,
                                                   chunk_index_meta_info.base_row_id_,
                                                   chunk_index_meta_info.row_cnt_,
                                                   chunk_index_meta_info.term_cnt_,
                                                   chunk_index_meta_info.base_name_,
                                                   chunk_index_meta_info.index_size_,
                                                   chunk_index_meta);
            if (!status.ok()) {
                return status;
            }
            memory_indexer.reset();
        }
    }
    mem_index->SetFulltextIndex(memory_indexer);
    return Status::OK();
}

Status NewTxn::PopulateIvfIndexInner(std::shared_ptr<IndexBase> index_base,
                                     SegmentIndexMeta &segment_index_meta,
                                     SegmentMeta &segment_meta,
                                     std::shared_ptr<ColumnDef> column_def,
                                     std::vector<ChunkID> &new_chunk_ids) {
    RowID base_row_id(segment_index_meta.segment_id(), 0);
    u32 row_count = 0;
    {
        auto [rc, status] = segment_meta.GetRowCnt1();
        if (!status.ok()) {
            return status;
        }
        row_count = rc;
    }
    Status status;
    ChunkID chunk_id = 0;
    std::tie(chunk_id, status) = segment_index_meta.GetAndSetNextChunkID();
    if (!status.ok()) {
        return status;
    }
    new_chunk_ids.push_back(chunk_id);
    std::optional<ChunkIndexMeta> chunk_index_meta;
    BufferObj *buffer_obj = nullptr;
    {
        Status status = NewCatalog::AddNewChunkIndex1(segment_index_meta,
                                                      this,
                                                      chunk_id,
                                                      base_row_id,
                                                      row_count,
                                                      0 /*term_count*/,
                                                      "" /*base_name*/,
                                                      0 /*index_size*/,
                                                      chunk_index_meta);
        if (!status.ok()) {
            return status;
        }
        status = chunk_index_meta->GetIndexBuffer(buffer_obj);
        if (!status.ok()) {
            return status;
        }
    }
    {
        BufferHandle buffer_handle = buffer_obj->Load();
        auto *data_ptr = static_cast<IVFIndexInChunk *>(buffer_handle.GetDataMut());
        data_ptr->BuildIVFIndex(segment_meta, row_count, column_def);
    }
    buffer_obj->Save();
    return Status::OK();
}

Status NewTxn::PopulateEmvbIndexInner(std::shared_ptr<IndexBase> index_base,
                                      SegmentIndexMeta &segment_index_meta,
                                      SegmentMeta &segment_meta,
                                      std::shared_ptr<ColumnDef> column_def,
                                      std::vector<ChunkID> &new_chunk_ids) {
    RowID base_row_id(segment_index_meta.segment_id(), 0);
    u32 row_count = 0;
    {
        auto [rc, status] = segment_meta.GetRowCnt1();
        if (!status.ok()) {
            return status;
        }
        row_count = rc;
    }
    ChunkID chunk_id = 0;
    Status status;
    std::tie(chunk_id, status) = segment_index_meta.GetAndSetNextChunkID();
    if (!status.ok()) {
        return status;
    }
    new_chunk_ids.push_back(chunk_id);
    std::optional<ChunkIndexMeta> chunk_index_meta;
    BufferObj *buffer_obj = nullptr;
    {
        Status status = NewCatalog::AddNewChunkIndex1(segment_index_meta,
                                                      this,
                                                      chunk_id,
                                                      base_row_id,
                                                      row_count,
                                                      0 /*term_count*/,
                                                      "" /*base_name*/,
                                                      0 /*index_size*/,
                                                      chunk_index_meta);
        status = chunk_index_meta->GetIndexBuffer(buffer_obj);
        if (!status.ok()) {
            return status;
        }
    }
    {
        BufferHandle buffer_handle = buffer_obj->Load();
        auto *data_ptr = static_cast<EMVBIndex *>(buffer_handle.GetDataMut());
        data_ptr->BuildEMVBIndex(base_row_id, row_count, segment_meta, column_def);
    }
    buffer_obj->Save();
    return Status::OK();
}

Status NewTxn::OptimizeFtIndex(std::shared_ptr<IndexBase> index_base,
                               SegmentIndexMeta &segment_index_meta,
                               RowID &base_rowid_out,
                               u32 &row_cnt_out,
                               u32 &term_cnt_out,
                               std::string &base_name_out) {
    const auto *index_fulltext = static_cast<const IndexFullText *>(index_base.get());

    std::vector<ChunkID> chunk_ids;
    {
        auto [chunk_ids_ptr, status] = segment_index_meta.GetChunkIDs1();
        if (!status.ok()) {
            return status;
        }
        chunk_ids = *chunk_ids_ptr;
    }
    if (chunk_ids.size() <= 1) {
        return Status::OK();
    }
    std::string msg = fmt::format("merging {}", *index_fulltext->index_name_);
    std::vector<std::string> base_names;
    std::vector<RowID> base_rowids;
    RowID base_rowid;
    u32 total_row_count = 0;
    u32 total_term_count = 0;
    std::vector<ChunkID> old_ids;

    for (auto iter = chunk_ids.begin(); iter != chunk_ids.end(); ++iter) {
        ChunkID chunk_id = *iter;
        ChunkIndexMeta chunk_index_meta(chunk_id, segment_index_meta);
        ChunkIndexMetaInfo *chunk_info_ptr = nullptr;
        {
            Status status = chunk_index_meta.GetChunkInfo(chunk_info_ptr);
            if (!status.ok()) {
                return status;
            }
        }
        // Initialize base_rowid for the first chunk
        if (iter == chunk_ids.begin()) {
            base_rowid = chunk_info_ptr->base_row_id_;
        }
        if (const RowID expect_base_row_id = base_rowid + total_row_count; chunk_info_ptr->base_row_id_ > expect_base_row_id) {
            msg += fmt::format(" stop at gap to chunk {}, expect_base_row_id: {:016x}, base_row_id: {:016x}",
                               chunk_info_ptr->base_name_,
                               expect_base_row_id.ToUint64(),
                               chunk_info_ptr->base_row_id_.ToUint64());
            chunk_ids.erase(iter, chunk_ids.end());
            break;
        } else if (chunk_info_ptr->base_row_id_ < expect_base_row_id) {
            msg += fmt::format(" found overlap to chunk {}, expect_base_row_id: {:016x}, base_row_id: {:016x}",
                               chunk_info_ptr->base_name_,
                               expect_base_row_id.ToUint64(),
                               chunk_info_ptr->base_row_id_.ToUint64());
            UnrecoverableError(msg);
            return Status::OK();
        }
        msg += " " + chunk_info_ptr->base_name_;
        base_names.push_back(chunk_info_ptr->base_name_);
        base_rowids.push_back(chunk_info_ptr->base_row_id_);
        total_row_count += chunk_info_ptr->row_cnt_;
        total_term_count += chunk_info_ptr->term_cnt_;
    }

    if (chunk_ids.size() <= 1) {
        msg += fmt::format(" skip merge due to only {} chunk", chunk_ids.size());
        LOG_INFO(msg);
        return Status::OK();
    }

    std::string dst_base_name = fmt::format("ft_{:016x}_{:x}", base_rowid.ToUint64(), total_row_count);
    msg += " -> " + dst_base_name;
    LOG_INFO(msg);

    std::shared_ptr<std::string> index_dir = segment_index_meta.GetSegmentIndexDir();
    ColumnIndexMerger column_index_merger(*index_dir, index_fulltext->flag_);
    column_index_merger.Merge(base_names, base_rowids, dst_base_name);
    {
        base_rowid_out = base_rowid;
        row_cnt_out = total_row_count;
        term_cnt_out = total_term_count;
        base_name_out = dst_base_name;
    }

    LOG_INFO(fmt::format("done merging {} {}", *index_fulltext->index_name_, dst_base_name));

    return Status::OK();
}

Status NewTxn::OptimizeVecIndex(std::shared_ptr<IndexBase> index_base,
                                std::shared_ptr<ColumnDef> column_def,
                                SegmentMeta &segment_meta,
                                RowID base_rowid,
                                u32 total_row_cnt,
                                BufferObj *buffer_obj) {
    auto [block_ids, status] = segment_meta.GetBlockIDs1();
    if (!status.ok()) {
        return status;
    }

    if (index_base->index_type_ == IndexType::kHnsw) {
        const auto *index_hnsw = static_cast<const IndexHnsw *>(index_base.get());
        if (index_hnsw->build_type_ == HnswBuildType::kLSG) {
            UnrecoverableError("Not implemented yet");
        }

        std::unique_ptr<HnswIndexInMem> memory_hnsw_index = HnswIndexInMem::Make(base_rowid, index_base.get(), column_def);
        for (BlockID block_id : *block_ids) {
            BlockMeta block_meta(block_id, segment_meta);
            size_t block_row_cnt = 0;
            std::tie(block_row_cnt, status) = block_meta.GetRowCnt1();
            if (!status.ok()) {
                return status;
            }
            ColumnMeta column_meta(column_def->id(), block_meta);
            size_t row_cnt = std::min(block_row_cnt, size_t(total_row_cnt));
            total_row_cnt -= row_cnt;
            ColumnVector col;
            status = NewCatalog::GetColumnVector(column_meta, column_meta.get_column_def(), row_cnt, ColumnVectorMode::kReadOnly, col);
            if (!status.ok()) {
                return status;
            }
            u32 offset = 0;
            memory_hnsw_index->InsertVecs(base_rowid.segment_offset_, col, offset, row_cnt);
        }

        memory_hnsw_index->Dump(buffer_obj);
    } else if (index_base->index_type_ == IndexType::kBMP) {
        auto memory_bmp_index = std::make_shared<BMPIndexInMem>(base_rowid, index_base.get(), column_def.get());

        for (BlockID block_id : *block_ids) {
            BlockMeta block_meta(block_id, segment_meta);
            size_t block_row_cnt = 0;
            std::tie(block_row_cnt, status) = block_meta.GetRowCnt1();
            if (!status.ok()) {
                return status;
            }
            ColumnMeta column_meta(column_def->id(), block_meta);
            size_t row_cnt = std::min(block_row_cnt, size_t(total_row_cnt));
            total_row_cnt -= row_cnt;
            ColumnVector col;
            status = NewCatalog::GetColumnVector(column_meta, column_meta.get_column_def(), row_cnt, ColumnVectorMode::kReadOnly, col);
            if (!status.ok()) {
                return status;
            }
            u32 offset = 0;
            memory_bmp_index->AddDocs(base_rowid.segment_offset_, col, offset, row_cnt);
        }
        memory_bmp_index->Dump(buffer_obj);
    } else {
        UnrecoverableError("Not implemented yet");
    }
    return Status::OK();
}

Status NewTxn::AlterSegmentIndexByParams(SegmentIndexMeta &segment_index_meta, const std::vector<std::unique_ptr<InitParameter>> &raw_params) {
    Status status;
    std::shared_ptr<IndexBase> index_base;

    std::tie(index_base, status) = segment_index_meta.table_index_meta().GetIndexBase();
    if (!status.ok()) {
        return status;
    }
    std::vector<ChunkID> *chunk_ids_ptr = nullptr;
    std::tie(chunk_ids_ptr, status) = segment_index_meta.GetChunkIDs1();
    if (!status.ok()) {
        return status;
    }
    std::shared_ptr<MemIndex> mem_index = segment_index_meta.GetMemIndex();

    switch (index_base->index_type_) {
        case IndexType::kBMP: {
            std::optional<BMPOptimizeOptions> ret = BMPUtil::ParseBMPOptimizeOptions(raw_params);
            if (!ret) {
                break;
            }
            const auto &options = ret.value();

            for (ChunkID chunk_id : *chunk_ids_ptr) {
                ChunkIndexMeta chunk_index_meta(chunk_id, segment_index_meta);
                BufferObj *index_buffer = nullptr;
                status = chunk_index_meta.GetIndexBuffer(index_buffer);
                if (!status.ok()) {
                    return status;
                }
                BufferHandle buffer_handle = index_buffer->Load();
                BMPHandlerPtr bmp_handler = *static_cast<BMPHandlerPtr *>(buffer_handle.GetDataMut());
                bmp_handler->Optimize(options);
            }
            std::shared_ptr<BMPIndexInMem> bmp_index = mem_index->GetBMPIndex();
            if (bmp_index) {
                BMPHandlerPtr bmp_handler = bmp_index->get();
                bmp_handler->Optimize(options);
            }

            break;
        }
        case IndexType::kHnsw: {
            auto params = HnswUtil::ParseOptimizeOptions(raw_params);
            if (!params) {
                break;
            }
            for (ChunkID chunk_id : *chunk_ids_ptr) {
                ChunkIndexMeta chunk_index_meta(chunk_id, segment_index_meta);
                BufferObj *index_buffer = nullptr;
                status = chunk_index_meta.GetIndexBuffer(index_buffer);
                if (!status.ok()) {
                    return status;
                }
                BufferHandle buffer_handle = index_buffer->Load();
                HnswHandlerPtr hnsw_handler = *static_cast<HnswHandlerPtr *>(buffer_handle.GetDataMut());
                if (params->compress_to_lvq) {
                    hnsw_handler->CompressToLVQ();
                } else if (params->compress_to_rabitq) {
                    hnsw_handler->CompressToRabitq();
                }
                if (params->lvq_avg) {
                    hnsw_handler->Optimize();
                }
            }
            if (mem_index) {
                std::shared_ptr<HnswIndexInMem> memory_hnsw_index = mem_index->GetHnswIndex();
                if (memory_hnsw_index) {
                    HnswHandlerPtr hnsw_handler = memory_hnsw_index->get();
                    if (params->compress_to_lvq) {
                        hnsw_handler->CompressToLVQ();
                    } else if (params->compress_to_rabitq) {
                        hnsw_handler->CompressToRabitq();
                    }
                    if (params->lvq_avg) {
                        hnsw_handler->Optimize();
                    }
                }
            }
            break;
        }
        default: {
            return Status::OK();
        }
    }

    return Status::OK();
}

Status NewTxn::ReplayAlterIndexByParams(WalCmdAlterIndexV2 *alter_index_cmd) {
    return AlterIndexByParams(alter_index_cmd->db_name_,
                              alter_index_cmd->table_name_,
                              alter_index_cmd->index_name_,
                              std::move(alter_index_cmd->params_));
}

Status NewTxn::DumpSegmentMemIndex(SegmentIndexMeta &segment_index_meta, const ChunkID &new_chunk_id) {
    std::shared_ptr<MemIndex> mem_index = segment_index_meta.PopMemIndex();
    if (mem_index == nullptr || (mem_index->GetBaseMemIndex() == nullptr && mem_index->GetEMVBIndex() == nullptr)) {
        return Status::EmptyMemIndex();
    }
    mem_index->WaitUpdate();
    LOG_TRACE(fmt::format("NewTxn::DumpSegmentMemIndex WaitUpdate mem_index {:p}", (void *)mem_index.get()));
    TableIndexMeta &table_index_meta = segment_index_meta.table_index_meta();
    auto [index_base, index_status] = table_index_meta.GetIndexBase();
    if (!index_status.ok()) {
        return index_status;
    }

    std::shared_ptr<SecondaryIndexInMem> memory_secondary_index = nullptr;
    std::shared_ptr<IVFIndexInMem> memory_ivf_index = nullptr;
    std::shared_ptr<HnswIndexInMem> memory_hnsw_index = nullptr;
    std::shared_ptr<BMPIndexInMem> memory_bmp_index = nullptr;
    std::shared_ptr<EMVBIndexInMem> memory_emvb_index = nullptr;

    // dump mem index only happens in parallel with read, not write, so no lock is needed.
    switch (index_base->index_type_) {
        case IndexType::kSecondary: {
            memory_secondary_index = mem_index->GetSecondaryIndex();
            if (memory_secondary_index == nullptr) {
                return Status::EmptyMemIndex();
            }
            break;
        }
        case IndexType::kFullText: {
            std::shared_ptr<MemoryIndexer> memory_indexer = mem_index->GetFulltextIndex();
            if (memory_indexer == nullptr) {
                return Status::EmptyMemIndex();
            }
            memory_indexer->Dump(false /*offline*/, false /*spill*/);
            break;
        }
        case IndexType::kIVF: {
            memory_ivf_index = mem_index->GetIVFIndex();
            if (memory_ivf_index == nullptr) {
                return Status::EmptyMemIndex();
            }
            break;
        }
        case IndexType::kHnsw: {
            memory_hnsw_index = mem_index->GetHnswIndex();
            if (memory_hnsw_index == nullptr) {
                return Status::EmptyMemIndex();
            }
            break;
        }
        case IndexType::kBMP: {
            memory_bmp_index = mem_index->GetBMPIndex();
            if (memory_bmp_index == nullptr) {
                return Status::EmptyMemIndex();
            }
            break;
        }
        case IndexType::kEMVB: {
            memory_emvb_index = mem_index->GetEMVBIndex();
            if (memory_emvb_index == nullptr) {
                return Status::EmptyMemIndex();
            }
            break;
        }
        case IndexType::kDiskAnn: {
            UnrecoverableError("Not implemented yet");
            break;
        }
        default: {
            UnrecoverableError("Not implemented yet");
        }
    }

    // NOTE: ChunkIndexMetaInfo::term_cnt_ is unstalbe before MemoryIndexer::Dump()!
    ChunkIndexMetaInfo chunk_index_meta_info;
    if (mem_index->GetBaseMemIndex() != nullptr) {
        chunk_index_meta_info = mem_index->GetBaseMemIndex()->GetChunkIndexMetaInfo();
    } else if (mem_index->GetEMVBIndex() != nullptr) {
        chunk_index_meta_info = mem_index->GetEMVBIndex()->GetChunkIndexMetaInfo();
    } else {
        UnrecoverableError("Invalid mem index");
    }

    std::optional<ChunkIndexMeta> chunk_index_meta;
    BufferObj *buffer_obj = nullptr;
    {
        Status status = NewCatalog::AddNewChunkIndex1(segment_index_meta,
                                                      this,
                                                      new_chunk_id,
                                                      chunk_index_meta_info.base_row_id_,
                                                      chunk_index_meta_info.row_cnt_,
                                                      chunk_index_meta_info.term_cnt_,
                                                      chunk_index_meta_info.base_name_,
                                                      chunk_index_meta_info.index_size_,
                                                      chunk_index_meta);
        if (!status.ok()) {
            return status;
        }

        chunk_infos_.push_back(ChunkInfoForCreateIndex{table_index_meta.table_meta().db_id_str(),
                                                       table_index_meta.table_meta().table_id_str(),
                                                       segment_index_meta.segment_id(),
                                                       new_chunk_id});

        status = chunk_index_meta->GetIndexBuffer(buffer_obj);
        if (!status.ok()) {
            return status;
        }
    }
    switch (index_base->index_type_) {
        case IndexType::kSecondary: {
            memory_secondary_index->Dump(buffer_obj);
            buffer_obj->Save();
            break;
        }
        case IndexType::kFullText: {
            break;
        }
        case IndexType::kIVF: {
            memory_ivf_index->Dump(buffer_obj);
            buffer_obj->Save();
            break;
        }
        case IndexType::kHnsw: {
            memory_hnsw_index->Dump(buffer_obj);
            buffer_obj->Save();
            if (buffer_obj->type() != BufferType::kMmap) {
                buffer_obj->ToMmap();
            }
            break;
        }
        case IndexType::kBMP: {
            memory_bmp_index->Dump(buffer_obj);
            buffer_obj->Save();
            if (buffer_obj->type() != BufferType::kMmap) {
                buffer_obj->ToMmap();
            }
            break;
        }
        case IndexType::kEMVB: {
            memory_emvb_index->Dump(buffer_obj);
            buffer_obj->Save();
            break;
        }
        default: {
            UnrecoverableError("Not implemented yet");
        }
    }

    mem_index->ClearMemIndex();
    auto *storage = InfinityContext::instance().storage();
    if (storage != nullptr) {
        auto *memindex_tracer = storage->memindex_tracer();
        if (memindex_tracer != nullptr) {
            memindex_tracer->DumpDone(mem_index);
        }
    }
    return Status::OK();
}

Status NewTxn::CountMemIndexGapInSegment(SegmentIndexMeta &segment_index_meta,
                                         SegmentMeta &segment_meta,
                                         std::vector<std::pair<RowID, u64>> &append_ranges) {
    Status status;
    std::vector<ChunkID> *chunk_ids_ptr = nullptr;
    std::tie(chunk_ids_ptr, status) = segment_index_meta.GetChunkIDs1();
    if (!status.ok()) {
        return status;
    }
    std::vector<ChunkIndexMetaInfo> chunk_index_meta_infos;
    for (ChunkID chunk_id : *chunk_ids_ptr) {
        ChunkIndexMeta chunk_index_meta(chunk_id, segment_index_meta);
        ChunkIndexMetaInfo *chunk_index_meta_info_ptr = nullptr;
        status = chunk_index_meta.GetChunkInfo(chunk_index_meta_info_ptr);
        if (!status.ok()) {
            return status;
        }
        chunk_index_meta_infos.push_back(*chunk_index_meta_info_ptr);
    }
    std::sort(chunk_index_meta_infos.begin(), chunk_index_meta_infos.end(), [](const ChunkIndexMetaInfo &lhs, const ChunkIndexMetaInfo &rhs) {
        return lhs.base_row_id_ < rhs.base_row_id_;
    });
    SegmentID segment_id = segment_meta.segment_id();

    RowID start_row_id(segment_id, 0);
    for (const ChunkIndexMetaInfo &chunk_index_meta_info : chunk_index_meta_infos) {
        if (chunk_index_meta_info.base_row_id_ != start_row_id) {
            UnrecoverableError(fmt::format("Chunk index row alignment error: Expected {}.{} but got {}.{}",
                                           start_row_id.segment_id_,
                                           start_row_id.segment_offset_,
                                           chunk_index_meta_info.base_row_id_.segment_id_,
                                           chunk_index_meta_info.base_row_id_.segment_offset_));
        }
        start_row_id = chunk_index_meta_info.base_row_id_ + chunk_index_meta_info.row_cnt_;
        LOG_TRACE(fmt::format("CountMemIndexGapInSegment: Chunk index row id {}.{}, row count {}",
                              chunk_index_meta_info.base_row_id_.segment_id_,
                              chunk_index_meta_info.base_row_id_.segment_offset_,
                              chunk_index_meta_info.row_cnt_));
    }

    std::vector<BlockID> *block_ids_ptr = nullptr;
    std::tie(block_ids_ptr, status) = segment_meta.GetBlockIDs1();
    if (!status.ok()) {
        return status;
    }
    size_t block_capacity = DEFAULT_BLOCK_CAPACITY;
    std::vector<BlockID> block_ids = *block_ids_ptr;
    sort(block_ids.begin(), block_ids.end());
    BlockID start_block_id = start_row_id.segment_offset_ / block_capacity;
    BlockOffset block_offset = start_row_id.segment_offset_ % block_capacity;
    {
        size_t i = start_block_id;
        if (i >= block_ids.size()) {
            return Status::OK();
        }
        for (; i < block_ids.size(); ++i) {
            BlockID block_id = block_ids[i];
            BlockMeta block_meta(block_id, segment_meta);
            size_t block_row_cnt = 0;
            std::tie(block_row_cnt, status) = block_meta.GetRowCnt1();
            if (!status.ok() || block_row_cnt == block_offset) {
                return status;
            }
            append_ranges.emplace_back(RowID(segment_id, (u32(block_id) << BLOCK_OFFSET_SHIFT) + block_offset), block_row_cnt - block_offset);
            block_offset = 0;
        }
    }
    return Status::OK();
}

Status NewTxn::RecoverMemIndex(TableIndexMeta &table_index_meta) {
    Status status;
    TableMeta &table_meta = table_index_meta.table_meta();

    std::vector<SegmentID> *segment_ids_ptr = nullptr;
    std::tie(segment_ids_ptr, status) = table_meta.GetSegmentIDs1();
    if (!status.ok()) {
        return status;
    }
    std::vector<SegmentID> *index_segment_ids_ptr = nullptr;
    std::tie(index_segment_ids_ptr, status) = table_index_meta.GetSegmentIndexIDs1();
    if (!status.ok()) {
        return status;
    }

    std::vector<std::pair<RowID, u64>> append_ranges;
    std::unordered_set<SegmentID> index_segment_ids_set(index_segment_ids_ptr->begin(), index_segment_ids_ptr->end());
    for (SegmentID segment_id : *segment_ids_ptr) {
        SegmentMeta segment_meta(segment_id, table_meta);
        if (!index_segment_ids_set.contains(segment_id)) {
            //            std::shared_ptr<std::string> error_msg = std::make_shared<std::string>(fmt::format("Segment {} not in index {}",
            //            segment_id, table_index_meta.index_id_str())); LOG_WARN(*error_msg); UnrecoverableError(*error_msg);//
            std::optional<SegmentIndexMeta> segment_index_meta;
            status = NewCatalog::AddNewSegmentIndex1(table_index_meta, this, segment_id, segment_index_meta);
            if (!status.ok()) {
                return status;
            }
            status = CountMemIndexGapInSegment(*segment_index_meta, segment_meta, append_ranges);
            if (!status.ok()) {
                return status;
            }
        } else {
            SegmentIndexMeta segment_index_meta(segment_id, table_index_meta);
            status = CountMemIndexGapInSegment(segment_index_meta, segment_meta, append_ranges);
            if (!status.ok()) {
                return status;
            }
        }
    }
    std::ostringstream oss;
    oss << "[";
    for (const auto &range : append_ranges) {
        oss << fmt::format("({}, {}), ", range.first.ToUint64(), range.second);
    }
    oss << "]";
    LOG_INFO(fmt::format("NewTxn::RecoverMemIndex db {} table {} index {} append_ranges {}: {}",
                         table_meta.db_id_str(),
                         table_meta.table_id_str(),
                         table_index_meta.index_id_str(),
                         append_ranges.size(),
                         oss.str()));
    for (const auto &range : append_ranges) {
        status = this->AppendIndex(table_index_meta, range);
        if (!status.ok()) {
            return status;
        }
    }
    return Status::OK();
}

Status NewTxn::GetFullTextIndexReader(const std::string &db_name, const std::string &table_name, std::shared_ptr<IndexReader> &index_reader) {
    std::shared_ptr<DBMeta> db_meta;
    std::shared_ptr<TableMeta> table_meta;
    TxnTimeStamp create_timestamp;
    Status status = GetTableMeta(db_name, table_name, db_meta, table_meta, create_timestamp);
    if (!status.ok()) {
        return status;
    }
    std::shared_ptr<TableIndexReaderCache> ft_index_cache;
    status = table_meta->GetFtIndexCache(ft_index_cache);
    if (!status.ok()) {
        return status;
    }
    index_reader = ft_index_cache->GetIndexReader(this);
    return Status::OK();
}

Status NewTxn::PrepareCommitCreateIndex(WalCmdCreateIndexV2 *create_index_cmd) {
    const TxnTimeStamp commit_ts = txn_context_ptr_->commit_ts_;
    const std::string &db_name = create_index_cmd->db_name_;
    const std::string &table_name = create_index_cmd->table_name_;
    const std::string &index_name = *create_index_cmd->index_base_->index_name_;
    const std::string &db_id_str = create_index_cmd->db_id_;
    const std::string &table_id_str = create_index_cmd->table_id_;
    const std::string &table_key = create_index_cmd->table_key_;
    const std::string &index_id_str = create_index_cmd->index_id_;
    std::shared_ptr<IndexBase> &index_base = create_index_cmd->index_base_;

    TableMeta table_meta(db_id_str, table_id_str, table_name, this);
    std::shared_ptr<TableIndexMeta> table_index_meta_ptr;
    Status status = new_catalog_->AddNewTableIndex(table_meta, index_id_str, commit_ts, index_base, table_index_meta_ptr);
    if (!status.ok()) {
        return status;
    }

    std::vector<SegmentID> *segment_ids_ptr = nullptr;
    std::tie(segment_ids_ptr, status) = table_meta.GetSegmentIDs1();
    if (!status.ok()) {
        return status;
    }

    if (this->IsReplay()) {
        WalCmdDumpIndexV2 dump_index_cmd(db_name, db_id_str, table_name, table_id_str, index_name, index_id_str, 0, table_key);
        dump_index_cmd.dump_cause_ = DumpIndexCause::kReplayCreateIndex;
        for (const WalSegmentIndexInfo &segment_index_info : create_index_cmd->segment_index_infos_) {
            dump_index_cmd.segment_id_ = segment_index_info.segment_id_;
            dump_index_cmd.chunk_infos_ = segment_index_info.chunk_infos_;

            status = this->ReplayDumpIndex(&dump_index_cmd);
            if (!status.ok()) {
                return status;
            }
        }
    } else {
        for (SegmentID segment_id : *segment_ids_ptr) {
            SegmentMeta segment_meta(segment_id, table_meta);
            size_t segment_row_cnt = 0;
            std::tie(segment_row_cnt, status) = segment_meta.GetRowCnt1();
            if (!status.ok()) {
                return status;
            }
            status = this->PopulateIndex(db_name,
                                         table_name,
                                         index_name,
                                         table_key,
                                         *table_index_meta_ptr,
                                         segment_meta,
                                         segment_row_cnt,
                                         DumpIndexCause::kCreateIndex,
                                         create_index_cmd);
            if (!status.ok()) {
                return status;
            }
        }
    }
    if (create_index_cmd->index_base_->index_type_ == IndexType::kFullText) {
        // Invalidate existing fulltext index cache
        status = table_meta.InvalidateFtIndexCache();
        if (!status.ok()) {
            return status;
        }

        LOG_TRACE(fmt::format("Created new fulltext index cache for index: {}", *create_index_cmd->index_base_->index_name_));
    }

    return Status::OK();
}

Status NewTxn::PrepareCommitDropIndex(const WalCmdDropIndexV2 *drop_index_cmd) {
    const std::string &db_id_str = drop_index_cmd->db_id_;
    const std::string &table_id_str = drop_index_cmd->table_id_;
    const std::string &table_name = drop_index_cmd->table_name_;
    const std::string &index_id_str = drop_index_cmd->index_id_;
    const std::string &index_name = drop_index_cmd->index_name_;
    const std::string &index_key = drop_index_cmd->index_key_;
    const TxnTimeStamp create_ts = infinity::GetTimestampFromKey(index_key);

    TxnTimeStamp commit_ts = txn_context_ptr_->commit_ts_;

    auto ts_str = std::to_string(commit_ts);
    kv_instance_->Put(KeyEncode::DropTableIndexKey(db_id_str, table_id_str, drop_index_cmd->index_name_, create_ts, index_id_str), ts_str);

    TableMeta table_meta(db_id_str, table_id_str, table_name, this);
    TableIndexMeta table_index_meta(index_id_str, index_name, table_meta);
    std::shared_ptr<IndexBase> index_base;
    Status status;
    std::tie(index_base, status) = table_index_meta.GetIndexBase();
    if (!status.ok()) {
        return status;
    }
    if (index_base->index_type_ == IndexType::kFullText) {
        // Invalidate fulltext index cache when dropping fulltext index
        std::shared_ptr<TableIndexReaderCache> ft_index_cache;
        Status cache_status = table_meta.GetFtIndexCache(ft_index_cache);
        if (cache_status.ok()) {
            ft_index_cache->Invalidate();
            LOG_TRACE(fmt::format("Invalidated fulltext index cache of table {} during drop index commit", drop_index_cmd->table_name_));
        }
    }

    return Status::OK();
}

Status NewTxn::PrepareCommitDumpIndex(const WalCmdDumpIndexV2 *dump_index_cmd, KVInstance *kv_instance) {
    const TxnTimeStamp commit_ts = txn_context_ptr_->commit_ts_;
    const std::string &db_id_str = dump_index_cmd->db_id_;
    const std::string &table_id_str = dump_index_cmd->table_id_;
    const std::string &table_name = dump_index_cmd->table_name_;
    const std::string &index_id_str = dump_index_cmd->index_id_;
    const std::string &index_name = dump_index_cmd->index_name_;
    SegmentID segment_id = dump_index_cmd->segment_id_;

    TableMeta table_meta(db_id_str, table_id_str, table_name, this);

    TableIndexMeta table_index_meta(index_id_str, index_name, table_meta);

    auto [index_base, status] = table_index_meta.GetIndexBase();
    if (!status.ok()) {
        return status;
    }
    if (index_base->index_type_ == IndexType::kFullText) {
        table_index_meta.table_meta().InvalidateFtIndexCache();
    }

    for (ChunkID deprecate_id : dump_index_cmd->deprecate_ids_) {

        auto ts_str = std::to_string(commit_ts);
        kv_instance_->Put(KeyEncode::DropChunkIndexKey(db_id_str, table_id_str, index_id_str, segment_id, deprecate_id), ts_str);
    }

    PersistenceManager *pm = InfinityContext::instance().persistence_manager();
    if (pm != nullptr) {
        // When all data and index is write to disk, try to finalize the
        PersistResultHandler handler(pm);
        PersistWriteResult result = pm->CurrentObjFinalize();
        handler.HandleWriteResult(result);
    }

    return Status::OK();
}

Status NewTxn::RestoreTableIndexesFromSnapshot(TableMeta &table_meta, const std::vector<WalCmdCreateIndexV2> &index_cmds, bool is_link_files) {
    // NewTxnManager *new_txn_manager = InfinityContext::instance().storage()->new_txn_manager();
    // new_txn_manager->PrintAllKeyValue();
    u64 max_index_id = 0;
    Status status;
    for (const auto &index_cmd : index_cmds) {
        u64 index_id = std::stoull(index_cmd.index_id_);
        if (index_id > max_index_id) {
            max_index_id = index_id;
        }
        const std::string &index_name = *(index_cmd.index_base_->index_name_);
        std::shared_ptr<TableIndexMeta> table_index_meta;
        if (!is_link_files) {
            status = new_catalog_->AddNewTableIndex(table_meta,
                                                    index_cmd.index_id_,
                                                    txn_context_ptr_->commit_ts_,
                                                    index_cmd.index_base_,
                                                    table_index_meta);
            if (!status.ok()) {
                return status;
            }
        } else {
            table_index_meta = std::make_shared<TableIndexMeta>(index_cmd.index_id_, index_name, table_meta);
        }

        for (const auto &segment_index : index_cmd.segment_index_infos_) {
            std::optional<SegmentIndexMeta> segment_index_meta;

            // Calculate next_chunk_id from existing chunk infos
            ChunkID next_chunk_id = 0;
            if (!segment_index.chunk_infos_.empty()) {
                next_chunk_id = std::max_element(segment_index.chunk_infos_.begin(),
                                                 segment_index.chunk_infos_.end(),
                                                 [](const WalChunkIndexInfo &a, const WalChunkIndexInfo &b) { return a.chunk_id_ < b.chunk_id_; })
                                    ->chunk_id_ +
                                1;
            }
            if (!is_link_files) {
                status = new_catalog_->RestoreNewSegmentIndex1(*table_index_meta, this, segment_index.segment_id_, segment_index_meta, next_chunk_id);
                if (!status.ok()) {
                    return status;
                }
            } else {
                segment_index_meta.emplace(segment_index.segment_id_, *table_index_meta);
            }

            for (const auto &chunk_index : segment_index.chunk_infos_) {
                std::optional<ChunkIndexMeta> chunk_index_meta;
                status = new_catalog_->RestoreNewChunkIndex1(*segment_index_meta,
                                                             this,
                                                             chunk_index.chunk_id_,
                                                             chunk_index.base_rowid_,
                                                             chunk_index.row_count_,
                                                             chunk_index.term_count_,
                                                             chunk_index.base_name_,
                                                             chunk_index.index_size_,
                                                             chunk_index_meta,
                                                             is_link_files);
                if (!status.ok()) {
                    return status;
                }
            }
        }
    }
    if (!is_link_files) {
        status = table_meta.SetNextIndexID(std::to_string(max_index_id + 1));
        if (!status.ok()) {
            return status;
        }
    }
    return Status::OK();
}

Status NewTxn::ManualDumpIndex(const std::string &db_name, const std::string &table_name) {

    Status status;

    // 1. Get table and index metadata
    std::shared_ptr<DBMeta> db_meta;
    std::shared_ptr<TableMeta> table_meta;
    std::shared_ptr<TableIndexMeta> table_index_meta;
    std::string table_key;
    std::string index_key;
    TxnTimeStamp create_timestamp;
    status = GetTableMeta(db_name, table_name, db_meta, table_meta, create_timestamp);
    if (!status.ok()) {
        return status;
    }
    // get all index ids
    std::vector<std::string> *index_ids_ptr = nullptr;
    std::vector<std::string> *index_names_ptr = nullptr;
    status = table_meta->GetIndexIDs(index_ids_ptr, &index_names_ptr);
    if (!status.ok()) {
        return status;
    }
    size_t index_count = index_ids_ptr->size();
    for (size_t i = 0; i < index_count; ++i) {
        const std::string &index_id_str = (*index_ids_ptr)[i];
        const std::string &index_name_str = (*index_names_ptr)[i];
        TableIndexMeta table_index_meta(index_id_str, index_name_str, *table_meta);

        // 2. Get all segment IDs for this index
        std::vector<SegmentID> *segment_ids_ptr = nullptr;
        std::tie(segment_ids_ptr, status) = table_index_meta.GetSegmentIndexIDs1();
        if (!status.ok()) {
            return status;
        }

        // 3. Loop through all segments and dump each one
        for (SegmentID segment_id : *segment_ids_ptr) {
            SegmentIndexMeta segment_index_meta(segment_id, table_index_meta);

            // 4. Get memory index for this segment
            std::shared_ptr<MemIndex> mem_index = segment_index_meta.GetMemIndex();
            if (mem_index == nullptr || mem_index->IsDumping() || (mem_index->GetBaseMemIndex() == nullptr && mem_index->GetEMVBIndex() == nullptr)) {
                LOG_INFO(fmt::format("Skipping segment {} - no memory index to dump", segment_id));
                continue;
            }
            mem_index->SetIsDumping(true);

            // 4.5. Additional check for EMVB index - ensure it's built before dumping

            // 5. Allocate new chunk ID for this dump
            ChunkID chunk_id = 0;
            Status status;
            std::tie(chunk_id, status) = segment_index_meta.GetAndSetNextChunkID();
            if (!status.ok()) {
                return status;
            }

            // 6. Get old chunk IDs for cleanup (if any exist)
            std::vector<ChunkID> old_chunk_ids;
            auto [existing_chunk_ids_ptr, chunk_status] = segment_index_meta.GetChunkIDs1();
            if (chunk_status.ok()) {
                old_chunk_ids = *existing_chunk_ids_ptr;
            }

            // 7. Actually dump the memory index to disk
            status = this->DumpSegmentMemIndex(segment_index_meta, chunk_id);
            if (!status.ok() && status.code() != ErrorCode::kEmptyMemIndex) {
                return status;
            }

            LOG_INFO(fmt::format("Successfully dumped segment {} to chunk {}", segment_id, chunk_id));
        }

        // 9. Update fulltext segment timestamp if needed
        // auto [index_base, index_status] = table_index_meta.GetIndexBase();
        // if (index_status.ok() && index_base->index_type_ == IndexType::kFullText) {
        //     TxnTimeStamp commit_ts = this->txn_context_ptr_->commit_ts_;
        //     status = table_index_meta.UpdateFulltextSegmentTS(commit_ts);
        //     if (!status.ok()) {
        //         return status;
        //     }
        // }
    }

    return Status::OK();
}

} // namespace infinity
